package com.rrd.dw.mr.all;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.HBaseUtil;

/**
 * 
 * @author liubaoxin
 *
 */
public class ExportALLTaobaoAddressDataToDW extends Configured implements Tool {
	public static final String CRT001 = "\001";
	public static final String[] TAOBAO_COLUMNS = { "nick_name", "real_name", "birthday", "sex", "login_name",
			"source_user_mobile", "source_user_idno", "source" };// taobao
	public static final String[] ADDRESS_COLUMNS = { "consignee", "province", "city", "full_address", "post_code",
			"mobile" };// taobao

	public static final String[] JOB_COLUMNS = { "finished:boolean", "requested:boolean", "successful:boolean",
			"start_time:date", "end_time:date", "system_id", "query", "try_times:int", "type", "message", "code" };// job

	@Override
	public int run(String[] args) throws Exception {
		Properties prop = new Properties();
		try (InputStream is = this.getClass().getClassLoader().getResourceAsStream("hbase.properties");) {
			prop.load(is);
		}

		Job job = Job.getInstance(HBaseConfiguration.create(this.getConf()));
		Configuration conf = job.getConfiguration();
		conf.set("hbase.zookeeper.quorum", prop.get("hbase.zookeeper.quorum").toString());
		conf.set("hbase.rootdir", prop.get("hbase.rootdir").toString());

		conf.set("mapreduce.map.memory.mb", "3072");
		conf.set("mapreduce.reduce.memory.mb", "5120");
		conf.set("hbase.client.scanner.timeout.period", "120");
		conf.set("mapreduce.output.fileoutputformat.compress", "true");
		conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
		conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.GzipCodec");
		conf.set("io.compression.codecs", "org.apache.hadoop.io.compress.GzipCodec");
		conf.set("mapreduce.task.io.sort.factor", "10");
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (otherArgs.length != 4) {
			System.err.println("Usage: DataExportTool <out> <start_time> <end_time>");
			return -1;
		}
		System.out.println("输出目录："+otherArgs[0]);
		Long startTime = HBaseUtil.parseDate(otherArgs[1]).getTime();
		Long endTime = HBaseUtil.parseDate(otherArgs[2]).getTime();
		String model = otherArgs[3];
		job.setJarByClass(ExportALLTaobaoAddressDataToDW.class);
		List<Scan> scans = new ArrayList<>();

		Scan scan = new Scan();
		scan.setCacheBlocks(false);
		scan.setCaching(200);
		scan.setMaxVersions(1);
		scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes("alitaobao_records"));
		if (!model.equals("ALL")) {
			System.out.println("增量运行.............");
			scan.setTimeRange(startTime, endTime + 14400000);
		} else {
			System.out.println("全量运行.............");

		}
		scans.add(scan);

		TableMapReduceUtil.initTableMapperJob(scans, ExportTaobaoDataToDWMapper.class, Text.class, Text.class, job);
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[0]));
		
		job.setReducerClass(ExportTaobaoDataToDWReducer.class);
		job.setNumReduceTasks(5);
		return job.waitForCompletion(true) ? 0 : 1;

	}

	public static void main(String[] args) throws Exception {
		Configuration conf = HBaseConfiguration.create();
		int res = ToolRunner.run(conf, new ExportALLTaobaoAddressDataToDW(), args);
		System.exit(res);
	}

	public static class ExportTaobaoDataToDWMapper extends TableMapper<Text, Text> {
		private long ts=DateUtil.formateToTimestamp("2018-05-12");
		@Override
		protected void map(ImmutableBytesWritable key, Result value, Context context)
				throws IOException, InterruptedException {
			Thread.sleep(20);
			StringBuffer sb = new StringBuffer("");
			Object[] rowkey = HBaseUtil.decodeRowkey(key.get());
			if (rowkey == null) {
				return;
			}
			Cell dataCell = null;
			Cell countCell = null;
			dataCell = value.getColumnLatestCell(Bytes.toBytes("job"), Bytes.toBytes("start_time"));
			long timestamp = dataCell==null?0:Long.parseLong(HBaseUtil.getValue("long",CellUtil.cloneValue(dataCell)).toString());
			if(timestamp>ts){
				return;
			}
			String jobid = rowkey[1].toString();
			
			dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes("source_user_idno"));
			String idno = dataCell == null ? "" : HBaseUtil.getValue(CellUtil.cloneValue(dataCell)).toString();
			dataCell = value.getColumnLatestCell(Bytes.toBytes("job"), Bytes.toBytes("system_id"));
			String system_id = dataCell == null ? "" : HBaseUtil.getValue(CellUtil.cloneValue(dataCell)).toString();
			
			sb.append(jobid).append(CRT001);
			// sb.append(jobid).append(CRT001).append(idno).append(timestampstr).append(CRT001);
			JSONObject jObj = new JSONObject();
			jObj.put("timestamp", timestamp);
			jObj.put("idNumber", idno);
			JSONObject taobao = new JSONObject();
			JSONObject address = null;
			JSONArray addressArray = new JSONArray();
			// 地址信息
			countCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes("AliTaobaoAddresscount"));
			if (countCell == null) {
				return;
			}

			int count = Bytes.toInt(CellUtil.cloneValue(countCell));
			for (int i = 0; i < count; i++) {
				address = new JSONObject();
				dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes("consignee" + "#" + i));
				address.put("consignee", dataCell == null ? "" : HBaseUtil.getValue(CellUtil.cloneValue(dataCell)));
				dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes("province" + "#" + i));
				address.put("province", dataCell == null ? "" : HBaseUtil.getValue(CellUtil.cloneValue(dataCell)));
				dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes("city" + "#" + i));
				address.put("city", dataCell == null ? "" : HBaseUtil.getValue(CellUtil.cloneValue(dataCell)));
				dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes("full_address" + "#" + i));
				address.put("fullAddress", dataCell == null ? "" : HBaseUtil.getValue(CellUtil.cloneValue(dataCell)));
				dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes("post_code" + "#" + i));
				address.put("postCode", dataCell == null ? "" : HBaseUtil.getValue(CellUtil.cloneValue(dataCell)));
				dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes("isDefault" + "#" + i));
				address.put("isDefault", dataCell == null ? "" : HBaseUtil.getValue("boolean",CellUtil.cloneValue(dataCell)));
				addressArray.add(address);

			}
			taobao.put("address", addressArray);
			jObj.put("taobao", taobao);
			sb.append(jObj.toJSONString()).append(CRT001).append(system_id);
			context.write(new Text("addresslist" + jobid), new Text(sb.toString()));
			sb = null;
		}

	}

	public static class ExportTaobaoDataToDWReducer extends Reducer<Text, Text, NullWritable, Text> {
		private static NullWritable key_null = NullWritable.get();
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			for(Text t : values){
        		context.write(key_null, t);
        	}
           
		}
	}
}
